package com.thering;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;

/**
 * author: the_ring
 * date: 2022/6/8
 * function: movie info
 */
public class MovieInfo {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        // 获取昨天
        Calendar cal = Calendar.getInstance();
        cal.add(Calendar.DATE, -1);
        Date date = cal.getTime();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
        String yesterday = simpleDateFormat.format(date);

        // 配置环境
        Configuration conf = new Configuration();
        conf.set("date", yesterday);
//        conf.set("fs.defaultFS", "hdfs://ring1:8020");
//        conf.set("hbase.rootdir", "hdfs://ring1:8020/hbase");
//        conf.set("hbase.zookeeper.quorum", "ring1:2181,ring2:2181,ring3:2181");
//        conf.set("hbase.client.retries.number", "3");

        // job 配置
        Job job = Job.getInstance(conf);
        job.setJobName("movie info change");
        job.setJarByClass(MovieInfo.class);

        // map 配置
        job.setMapperClass(MovieInfoMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        TableMapReduceUtil.initTableReducerJob(
                "movie_info_change",
                MovieInfoReducer.class,
                job
        );

        FileInputFormat.addInputPath(job, new Path("/project/" + yesterday + "/movies/part-m*"));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
